package com.vtron.mq.service.impl;

import com.alibaba.fastjson.JSON;
import com.vtron.mq.models.MsgPackage;
import com.vtron.mq.service.ICommandService;
import com.vtron.mq.service.IMqListenerService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;

import java.util.Map;
import java.util.Objects;

/**
 * 命令处理器
 * @date : 2022/4/26 18:00
 * @author: 忻龙彪
 */
@Service
public class CommandListenerServiceImpl implements IMqListenerService {

    private final Logger logger = LoggerFactory.getLogger(CommandListenerServiceImpl.class);

    @Autowired(required = false)
    Map<String, ICommandService> commandServicesMap;

    @Override
    @KafkaListener(topics = "cmd")
    public void received(ConsumerRecord<String, String> record, Acknowledgment ack) {
        String value = record.value();
        logger.info("received package data:" + value);
        MsgPackage msg = JSON.parseObject(value, MsgPackage.class);
        if (!Objects.isNull(commandServicesMap)
                && commandServicesMap.containsKey(msg.getName())) {
            commandServicesMap.get(msg.getName()).execute(msg);
        }
        //手动提交offset
        ack.acknowledge();
    }
}
